package com.chenjj.bigdata.flink.stream.indicators;

import cn.hutool.json.JSONUtil;
import com.chenjj.bigdata.flink.stream.indicators.vo.SummaryByQD;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * 指标汇总
 *
 * 数据源为2个文件: request.txt  , response.txt
 *
 * 对这个2个文件进行指标计算
 */
public class IndicatorsSummary {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);//并行度，和生成的文件数是一致的

        // get input data by connecting to the socket
        DataStream<String> reqDS = env.readTextFile("E:\\IntelliJWorkspace\\gitee\\bigdata\\flink\\src\\main\\resources\\indicators\\request.txt");
        DataStream<String> repDS = env.readTextFile("E:\\IntelliJWorkspace\\gitee\\bigdata\\flink\\src\\main\\resources\\indicators\\response.txt");

        DataStream<SummaryByQD>  dataStream = reqDS.flatMap(new FlatMapFunction<String, SummaryByQD>() {
            @Override
            public void flatMap(String s, Collector<SummaryByQD> collector) throws Exception {
                String [] lines = s.split("\n");
                SummaryByQD vo;
                for (String line: lines) {
                    vo = JSONUtil.parseObj(line).toBean(SummaryByQD.class);
                    collector.collect(vo);
                }
            }
        }).keyBy("jyqd")
  //      .timeWindow(Time.seconds(10),Time.seconds(10))
        .reduce(new ReduceFunction<SummaryByQD>() {
            @Override
            public SummaryByQD reduce(SummaryByQD t0, SummaryByQD t1) throws Exception {
                 t1.setCount(t0.getCount()+1);
                t1.setJyje(t0.getJyje() + t1.getJyje());
                return t1;
            }
        });


        //把数据打印到控制台,使用一个并行度
        dataStream.writeAsText("D:\\tmp\\flink",FileSystem.WriteMode.OVERWRITE);

        //注意：因为flink是懒加载的，所以必须调用execute方法，上面的代码才会执行
        env.execute("streaming IndicatorsSummary");

       // Thread.sleep(50000);
    }
}
